package com.amigo.online.provider.kafka.receiver;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

/**
 * @version: V 1.0.0
 * @Description: （）
 * @author: zxy  QQ:1363320658
 * @date: 2018年12月13日 上午11:58:15
 */

@Component
public class KakfaDataComparisonFailLogReceiver {

    private static Logger logger = LoggerFactory.getLogger(KafkaMysqlBinlogReceiver.class);

    @KafkaListener(id = "comaprfail", topics = {"comparsionMongoUser"})
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info(message.toString());
            JSONObject faillog = JSON.parseObject(message.toString());
        }
    }
	
/*	@KafkaListener(id = "comaprfail", topics = { "comparsionMongoUser" })
	public void listenTo(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			Object message = kafkaMessage.get();
			logger.info(message.toString());
			JSONObject faillog = JSON.parseObject(message.toString());		
		}
	}
	
	@KafkaListener(id = "comaprfail", topics = { "comparsionMongoUser" })
	public void listen(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			Object message = kafkaMessage.get();
			logger.info(message.toString());
			JSONObject faillog = JSON.parseObject(message.toString());		
		}
	}
	
	@KafkaListener(id = "comaprfail", topics = { "comparsionMongoUser" })
	public void listen(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			Object message = kafkaMessage.get();
			logger.info(message.toString());
			JSONObject faillog = JSON.parseObject(message.toString());		
		}
	}
	
	@KafkaListener(id = "comaprfail", topics = { "comparsionMongoUser" })
	public void listen(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			Object message = kafkaMessage.get();
			logger.info(message.toString());
			JSONObject faillog = JSON.parseObject(message.toString());		
		}
	}*/
}
